/*
 * c_array_queue.c
 *
 *  Created on: Mar 8, 2012
 *      Author: lin
 */

#include <assert.h>
#include "zmalloc/zmalloc.h"
#include "concurrent.h"
#include "c_array_queue.h"

c_array_queue_t* c_array_queue_new0(uint32_t max_size,
        c_array_queue_ele_free_proc ele_del) {
    uint32_t max = 1;
    while (max < max_size)
        max <<= 1;
    c_array_queue_t* queue = zcalloc(
            sizeof(c_array_queue_t) + sizeof(complex_ele_t) * max);
    queue->max_size = max;
    queue->size_mask = max - 1;
    queue->ele_del = ele_del;
    spin_init(&queue->spin);

    return queue;
}

c_array_queue_t* c_array_queue_new1(uint32_t max_size) {
    return c_array_queue_new0(max_size, NULL);
}

static inline int _c_array_queue_eq_ele(c_array_queue_t* queue,
        complex_ele_t ele) {
    spin_lock(&queue->spin);
    if (queue->size == queue->max_size) {
        spin_unlock(&queue->spin);
        return 0;
    }
    queue->array[queue->eq_index++] = ele;
    queue->eq_index &= queue->size_mask;
    atomic_inc_get(&queue->size);
    spin_unlock(&queue->spin);
    return 1;
}

static inline int _c_array_queue_dq_ele(c_array_queue_t* queue,
        complex_ele_t *ele) {
    spin_lock(&queue->spin);
    if (queue->size == 0) {
        spin_unlock(&queue->spin);
        return 0;
    }
    *ele = queue->array[queue->dq_index++];
    queue->dq_index &= queue->size_mask;
    atomic_dec_get(&queue->size);
    spin_unlock(&queue->spin);
    return 1;
}

int c_array_queue_eq(c_array_queue_t* queue, const void* ele) {
    complex_ele_t cele;
    cele.ptr = (void*) ele;
    return _c_array_queue_eq_ele(queue, cele);

}
int c_array_queue_eq_s64(c_array_queue_t* queue, int64_t ele) {
    complex_ele_t cele;
    cele.s64 = ele;
    return _c_array_queue_eq_ele(queue, cele);
}
int c_array_queue_eq_u64(c_array_queue_t*queue, uint64_t ele) {
    complex_ele_t cele;
    cele.u64 = ele;
    return _c_array_queue_eq_ele(queue, cele);
}

void* c_array_queue_dq(c_array_queue_t* queue) {
    complex_ele_t ele;
    if (_c_array_queue_dq_ele(queue, &ele)) {
        return ele.ptr;
    }
    return NULL;
}
int c_array_queue_dq_s64(c_array_queue_t* queue, int64_t *s64) {
    complex_ele_t ele;
    if (_c_array_queue_dq_ele(queue, &ele)) {
        *s64 = ele.s64;
        return 1;
    }
    return 0;
}
int c_array_queue_dq_u64(c_array_queue_t* queue, uint64_t *u64) {
    complex_ele_t ele;
    if (_c_array_queue_dq_ele(queue, &ele)) {
        *u64 = ele.u64;
        return 1;
    }
    return 0;
}

void c_array_queue_free(c_array_queue_t* queue) {
    c_array_queue_ele_free_proc ele_del = queue->ele_del;
    if (ele_del) {
        void* ele = c_array_queue_dq(queue);
        while (ele) {
            ele_del(ele);
        }
    }
    spin_destory(&queue->spin);
    zfree(queue);
}

